
package mvcc

import (
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/pkg/v3/traceutil"
	"go.etcd.io/etcd/server/v3/lease"
	"go.etcd.io/etcd/server/v3/storage/backend"
	"go.etcd.io/etcd/server/v3/storage/schema"

	"go.uber.org/zap"
)

// non const可通过测试修改
var (
	// chanBufLen是用于发送监视事件的缓冲chan 
	// 的长度。
	// 请参阅https:
	chanBufLen = 128

	// MaxWatcherPersync是单个批中要同步的观察程序数
	maxWatchersPerSync = 512
)

type watchable interface {
	watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse, fcs ...FilterFunc) (*watcher, cancelFunc)
	progress(w *watcher)
	rev() int64
}

// watchableStore结构体完成了注册watcher实例、管理watcher实例，
// 以及发送触发watcher之后的响应等核心功能。在结构体watchableStore中内嵌了store，
type watchableStore struct {
	*store
	// 在修改Synced watcherGroup、unsynced watcherGroup等字段时，需要获取该锁进行同步。
	mu sync.RWMutex
	// 如果watcher实例关联的ch通道被阻塞了，则对应的watcherBatch实例会暂时记录到该字段中。
	// 在watchableStore实例中还会启动一个后台goroutine来处理该字段中保存的watcherBatch实例。
	// watcherBatch实际上就是map[*watcher] *eventBatch类型，watcherBatch只提供了一个add（）方法，
	// 用来添加被触发的watcher实例与其eventBatch之间的映射关系。
	victims []watcherBatch
	// 当有新的watcherBatch实例添加到victims字段中时，会向该通道中发送一个空结构体作为信号。
	victimc chan struct{}
	// 在结构体watchableStore中包含两个watcherGroup实例，即synced字段和unsynced宇段。
	// syncedwatcherGroup中全部的watch巳r实例都已经同步完毕，
	// 并等待新的更新操作：syncedwatcherGroup中的watcher实例都落后于当前最新更新操作，
	// 并且有一个单独的后台goroutine帮助其进行追赶。当etcd服务端收到客户端的watch请求时，
	// 如果请求携带了revision参数，
	// 则比较该请求的revision信息和store.currentRev信息：如果请求中的revision信息较大，
	// 则放入syncedwatcherGroup中，否则放入unsyncedwatcherGroup。在后面的分析中会看到，
	// watchableStore实例会启动一个后台的goroutine持续同步unsyncedwatcherGroup，
	// 然后将完成同步的watch巳r实例迁移到syncedwatcherGroup中储存。
	unsynced watcherGroup
	synced watcherGroup
	stopc chan struct{}
	// 在watchableStore实例中会启动两个后台goroutine，
	// 在watchableStore.Close（）方法中会通过该sync.WaitGroup实例实现等待两个后台goroutine 执行完成的功能。
	wg    sync.WaitGroup
}

// cancelFunc在运行
// 取消操作时更新未同步和已同步的映射。
type cancelFunc func()

func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) WatchableKV {
	return newWatchableStore(lg, b, le, cfg)
}

// newWatchableStore（）函数完成了watchableStore实例的初始化，其中会启动两个后台goroutine，
func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) *watchableStore {
	if lg == nil {
		lg = zap.NewNop()
	}
	s := &watchableStore{
		store:    NewStore(lg, b, le, cfg),	// 创建store实例
		victimc:  make(chan struct{}, 1),
		unsynced: newWatcherGroup(),	// 初始化unsynced watcherGroup
		synced:   newWatcherGroup(),	// 初始化synced watcherGroup
		stopc:    make(chan struct{}),
	}
	// 创建readView和writeView实例
	s.store.ReadView = &readView{s}
	s.store.WriteView = &writeView{s}
	if s.le != nil {
		// 将此存储用作删除程序，以便撤销触发监视事件
		s.le.SetRangeDeleter(func() lease.TxnDelete { return s.Write(traceutil.TODO()) })
	}
	s.wg.Add(2)	// 初始化sync.WaitGroup实例
	go s.syncWatchersLoop()	// 启动处理unsynced watcherGroup的后台goroutine
	go s.syncVictimsLoop()	// 启动处理victims的后台goroutine
	return s
}

func (s *watchableStore) Close() error {
	close(s.stopc)
	s.wg.Wait()
	return s.store.Close()
}

func (s *watchableStore) NewWatchStream() WatchStream {
	watchStreamGauge.Inc()
	return &watchStream{
		watchable: s,
		ch:        make(chan WatchResponse, chanBufLen),
		cancels:   make(map[WatchID]cancelFunc),
		watchers:  make(map[WatchID]*watcher),
	}
}

// 添加watcher实例的功能，该功能是在其watch（）方法中实现
func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse, fcs ...FilterFunc) (*watcher, cancelFunc) {
	wa := &watcher{	// 创建watcher实例
		key:    key,
		end:    end,
		minRev: startRev,
		id:     id,
		ch:     ch,
		fcs:    fcs,
	}
	// 加mu锁，因为要读取currentRev字段，所以还需要加revMu锁
	s.mu.Lock()
	s.revMu.RLock()
	// 比较startRev与store.currentRev，决定待添加的watcher实例是否已同步完成
	synced := startRev > s.store.currentRev || startRev == 0
	if synced {
		wa.minRev = s.store.currentRev + 1	// 设置待添加watcher的minRev字段值
		if startRev > wa.minRev {
			wa.minRev = startRev
		}
		// 如果待添加watcher已同步完成，则将其添加到synced watcherGroup中，
		s.synced.add(wa)
	} else {
		slowWatcherGauge.Inc()
		// 如果待添加watcher未同步完成，则将其添加到unsynced watcherGroup中，
		s.unsynced.add(wa)
	}
	s.revMu.RUnlock()
	s.mu.Unlock()

	watcherGauge.Inc()
	// 返回该watcher实例，以及取消该watcher实例的回调函数
	return wa, func() { s.cancelWatcher(wa) }
}

// watchableStore.watch（）方法返回的取消watcher实例的回调函数是通过调用watchableStore.cancelWatcher（）方法实现的
func (s *watchableStore) cancelWatcher(wa *watcher) {
	for {
		s.mu.Lock()
		if s.unsynced.delete(wa) {	// 尝试从unsynced watcherGroup中删除该watcher实例
			slowWatcherGauge.Dec()
			watcherGauge.Dec()
			break
		} else if s.synced.delete(wa) {	// ／尝试从synced watcherGroup中删除该watcher实例
			watcherGauge.Dec()
			break
		} else if wa.compacted {	// 如果待删除的watcher实例已经因为压缩操作而删除，则直接返回
			watcherGauge.Dec()
			break
		} else if wa.ch == nil {
			// 已取消（例如，取消/关闭竞争）
			break
		}

		if !wa.victim {
			s.mu.Unlock()
			panic("watcher not victim but not in watch groups")
		}
		// 如果在（un)synced watcherGroup中都没有，该watcher可能已经被触发了，则在victims中查找
		var victimBatch watcherBatch
		for _, wb := range s.victims {
			if wb[wa] != nil {	// 查找待删除watcher对应的watcherBatch实例
				victimBatch = wb
				break
			}
		}
		if victimBatch != nil {
			slowWatcherGauge.Dec()
			watcherGauge.Dec()
			// 删除victims字段中对应的watcher实例
			delete(victimBatch, wa)
			break
		}

		// 如果未找到，可能是待删除的watcher刚刚从synced watcherGroup中删除且未添加到victims中，所以稍后进行重试
		s.mu.Unlock()
		time.Sleep(time.Millisecond)
	}

	wa.ch = nil
	s.mu.Unlock()
}

func (s *watchableStore) Restore(b backend.Backend) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	err := s.store.Restore(b)
	if err != nil {
		return err
	}

	for wa := range s.synced.watchers {
		wa.restore = true
		s.unsynced.add(wa)
	}
	s.synced = newWatcherGroup()
	return nil
}

// 在初始化watchableStore实例时会启动两个后台gorouti时，其中一个goroutine执行watchableStore.
// sync WatchersLoop（）方法，该goroutine的主要作用是每隔1OOms对unsynced watcherGroup进行一次批量的同步
func (s *watchableStore) syncWatchersLoop() {
	defer s.wg.Done()	// 该后台goroutine结束时调用，

	for {
		s.mu.RLock()
		st := time.Now()
		lastUnsyncedWatchers := s.unsynced.size()	// 获取当前的unsynced watcherGroup的大小
		s.mu.RUnlock()

		unsyncedWatchers := 0
		if lastUnsyncedWatchers > 0 {
			// 存在需要进行同步的watcher实例，则调用syncWatchers（）方法对unsynced watcherGroup中的watcher进行批量同步
			unsyncedWatchers = s.syncWatchers()
		}
		syncDuration := time.Since(st)

		waitDuration := 100 * time.Millisecond
		// 调整该后台goroutine执行的时间间隔
		if unsyncedWatchers != 0 && lastUnsyncedWatchers > unsyncedWatchers {
			// 公平对待其他存储操作，允许花费时间
			waitDuration = syncDuration
		}

		select {	// 等待waitDuration时长之后，再开始下次同步
		case <-time.After(waitDuration):
		case <-s.stopc:
			return
		}
	}
}

// 在初始化 watchableStore 实例时启动的另一个后台 goroutine 会执行 syncVictimsLoop（）方法，
// 在该方法中会定期处理 watchableStore.victims 中 缓存的 watcherBatch 实例。
func (s *watchableStore) syncVictimsLoop() {
	defer s.wg.Done()	// 该后台 go routine 结束时调用，
	for {
		for s.moveVictims() != 0 {	// 循环处理 victims 中缓存的 watcherBatch 实例
		}
		s.mu.RLock()
		isEmpty := len(s.victims) == 0
		s.mu.RUnlock()

		var tickc <-chan time.Time
		if !isEmpty {
			tickc = time.After(10 * time.Millisecond)
		}
		// 根据 victims 的长度决定下次处理 victims 的时间间 隔
		select {
		case <-tickc:
		case <-s.victimc:
		case <-s.stopc:
			return
		}
	}
}

// watchableStore.move Victims（）方法是处理 victims 的核心，该方法会遍历 victims 宇段中记录
// 的 watchBatch 实例，并尝试将其中的 Event 实例封装成 watchResponse 重新发送 。 如果发送依
// 然失败，则将其放回 victims 宇段中保存，等待下一次重试；如果发送成功， 则根据相应的 watcher
// 的同步情况，将 watcher 实例迁移到（ un ) synced watcherGroup 中 。
func (s *watchableStore) moveVictims() (moved int) {
	s.mu.Lock()
	victims := s.victims
	s.victims = nil	// 清空 victims 字段
	s.mu.Unlock()

	var newVictim watcherBatch
	for _, wb := range victims {	// 遍历 victims 中记录的 watcherBatch
		// 再次尝试发送响应
		for w, eb := range wb {	// 从 watcherBatch 中获取 eventBatch 实例
			// 观察者已观察到存储，但不包括w.minRev 
			rev := w.minRev - 1
			// 将 eventBatch 封装成 watchResponse ，并调用 watcher.send（）方法尝试将其发送出去
			if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: rev}) {
				pendingEventsGauge.Add(float64(len(eb.evs)))
			} else {	// 如果 watcher.ch 通道依然阻塞，则将对应的 Event 实例重新放回 newVictim 变量中保存
				if newVictim == nil {
					newVictim = make(watcherBatch)
				}
				newVictim[w] = eb
				continue
			}
			moved++	// 记录重试成功的 eventBatch 个数
		}

		// 将已完成的受害者观察者分配给不同步/同步
		s.mu.Lock()
		s.store.revMu.RLock()
		curRev := s.store.currentRev
		for w, eb := range wb {	// 遍历 watcherBatch
			// 如果 eventBatch 实例被记录到 newVictim 中，则表示 watcher.ch 通过依然阻塞，
			// watchResponse 发送失败
			if newVictim != nil && newVictim[w] != nil {
				continue
			}
			// 如果 eventBatch 实例未被记录到 newVictim 中，则表示 watchResponse 发送成功
			w.victim = false
			if eb.moreRev != 0 {	// 检测当前 watcher 后续是否还有未同步的 Event 事件
				w.minRev = eb.moreRev
			}
			if w.minRev <= curRev {	// 当前 watcher 未完成同步，移动到 unsynced watcherGroup 中
				s.unsynced.add(w)
			} else {	// 当前 watcher 已经完成同步，移动到 synced watcherGroup 中
				slowWatcherGauge.Dec()
				s.synced.add(w)
			}
		}
		s.store.revMu.RUnlock()
		s.mu.Unlock()
	}

	if len(newVictim) > 0 {	// 使用 newVictim 更新 watchableStore.victims 字段
		s.mu.Lock()
		s.victims = append(s.victims, newVictim)
		s.mu.Unlock()
	}

	return moved	// 返回重试成功 的 eventBatch 个数
}

// syncWatchers（）方法是批量同步unsynced watcherGroup的核心，该方法的大致步骤如下：
// (1）从unsynced watcherGroup中选择一批watcher实例，作为此次需要进行同步的watcher实例。
// (2）从该批watcher实例中查找最小的minRev字段值。
// (3）在BoltDB中进行范围查询，查询minRev～currentRev（当前revision值）的所有键值对。
// (4）过滤掉该批watcher实例中因minRev～currentRev之间发生压缩操作而被清除的watcher实例。
// (5）遍历步骤3中查询到的键值对，将其中的更新操作转换为Event事件，然后封装成WatchResponse，并写入对应的watcher.ch通道中。
//     如果watcher.ch通道被填充满，则将Event事件记录到watchableStore.victims中，
//     并由另一个后台goroutine处理该字段中积累的Event事件。
// (6）将已经完成同步的watcher实例记录到synced watcherGroup中，同时将其中unsynced watcherGroup中删除。
func (s *watchableStore) syncWatchers() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	// 如果unsyncedwatcherGroup为空，则直接返回（
	if s.unsynced.size() == 0 {
		return 0
	}

	s.store.revMu.RLock()
	defer s.store.revMu.RUnlock()
	curRev := s.store.currentRev
	compactionRev := s.store.compactMainRev
	// 从unsynced watcherGroup中查找一批此次需要同步的watcher实例，
	// 并将这些watcher实例封装成watcherGroup实例返回，
	// 返回的minRev是这批待同步的 watcher实例中minRev字段最小值
	wg, minRev := s.unsynced.choose(maxWatchersPerSync, curRev, compactionRev)
	// 将minRev和currentRev写入minBytes和maxBytes中
	minBytes, maxBytes := newRevBytes(), newRevBytes()
	revToBytes(revision{main: minRev}, minBytes)
	revToBytes(revision{main: curRev + 1}, maxBytes)
	tx := s.store.b.ReadTx()	// 获取只读事务
	tx.RLock()
	// 调用UnsafeRange（）方法，对key Bucket进行范围查找
	revs, vs := tx.UnsafeRange(schema.Key, minBytes, maxBytes, 0)
	// 将查询到的键值对转换成对应的Event事件，
	evs := kvsToEvents(s.store.lg, wg, revs, vs)
	tx.RUnlock()

	victims := make(watcherBatch)
	// 将上述watcher集合及Event事件封装成watcherBatch
	wb := newWatcherBatch(wg, evs)
	for w := range wg.watchers {
		// 更新watcher.minRev字段，这样就不会被同一个修改操作触发两次
		w.minRev = curRev + 1

		eb, ok := wb[w]
		if !ok {
			// 该watcher实例所监听的键位对没有关新操作（在minRev～currentRev之间），
			// 所以没有被触发，故同步完毕，这里会将其转移到unsynced watcherGroup中
			s.synced.add(w)
			s.unsynced.delete(w)
			continue
		}

		if eb.moreRev != 0 {
			// 在minRev～currentRev之间触发当前watcher的更新操作过多，
			// 无法全部放到一个eventBatch中，这里只将该watcher.minRev设立成moreRev，
			// 则下次处理unsynced watcherGroup时，会将其后的更新操作查询出来继续处理
			w.minRev = eb.moreRev
		}
		// 将前面创建的Event事件集合封装成WatchResponse实例，然后写入watcher.ch通道中
		if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: curRev}) {
			pendingEventsGauge.Add(float64(len(eb.evs)))
		} else {	// 如果watcher.ch通过阻塞，则将该watcher.victim字段设置为true
			w.victim = true
		}
		// 如果watcher.ch迢迢阻塞，则将触发它的Event事件记录到victims中
		if w.victim {
			victims[w] = eb
		} else {
			// 如果后续还有其他未处理的Event事件，则当前watcher依然未同步完成
			if eb.moreRev != 0 {
				continue
			}
			// 当前watcher已经同步完成，加入synced watcherGroup中
			s.synced.add(w)
		}
		// 将当前watcher从unsynced watcherGroup中删除
		s.unsynced.delete(w)
	}
	// 将victims变量中记录watcherBatch添加到watchableStore.victims中，等待处理
	s.addVictim(victims)

	vsz := 0
	for _, v := range s.victims {
		vsz += len(v)
	}
	slowWatcherGauge.Set(float64(s.unsynced.size() + vsz))
	// 返回此次处理之后unsynced watcherGroup的长度
	return s.unsynced.size()
}

// kvsToEvents()函数，它负责将前面从 BoltDB 中查询到的键值对信息转换成相应的 Event 实例
func kvsToEvents(lg *zap.Logger, wg *watcherGroup, revs, vals [][]byte) (evs []mvccpb.Event) {
	for i, v := range vals {	// 键值对数据
		var kv mvccpb.KeyValue
		if err := kv.Unmarshal(v); err != nil {	// 反序列化得到 KeyValue 实例
			lg.Panic("failed to unmarshal mvccpb.KeyValue", zap.Error(err))
		}
		// 在此批处理的所有 watcher 实例中，是否有监听了该 Key 的实例，
		// 需要注意的是， unsynced watcherGroup 中可能同时记录了监听单个 Key 的 watcher 和监听范围的 watcher ，
		// 所以在 watcherGroup . contain （）方法中会同时查找这两种
		if !wg.contains(string(kv.Key)) {
			continue
		}

		ty := mvccpb.PUT	// 默认是更新操作
		if isTombstone(revs[i]) {
			ty = mvccpb.DELETE	// 如果是 tombstone ，则表示删除操作
			kv.ModRevision = bytesToRev(revs[i]).main
		}
		// 将该键位对转换成对应的 Event 实例 ， 并记录下来
		evs = append(evs, mvccpb.Event{Kv: &kv, Type: ty})
	}
	return evs
}

// watchableStore.notify（）方法是发送 Event 事件
func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) {
	victim := make(watcherBatch)
	// 将传入的 Event 实例转换成 watcerBatch(map[*watcher]eventBatch），之后进行遍历，
	// 逐个 watcher 进行处理
	for w, eb := range newWatcherBatch(&s.synced, evs) {
		if eb.revs != 1 {
			s.store.lg.Panic(
				"unexpected multiple revisions in watch notification",
				zap.Int("number-of-revisions", eb.revs),
			)
		}
		// 将当前 watcher 对应的 Event 实例封装成 watchResponse 实例 ，并尝试写入 watcher.ch 通道中
		if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: rev}) {
			pendingEventsGauge.Add(float64(len(eb.evs)))
		} else {	// 如果 wathcer.ch 通过阻塞，则将这些 Event 实例记录到 watchableStore.victims 字段中
			w.minRev = rev + 1
			w.victim = true
			victim[w] = eb
			s.synced.delete(w)	// 将该 watcher 从 synced watcherGroup 中删除
			slowWatcherGauge.Inc()
		}
	}
	s.addVictim(victim)
}

func (s *watchableStore) addVictim(victim watcherBatch) {
	if len(victim) == 0 {
		return
	}
	// 将 watcherBatch 实例追加到 watchableStore. victims 字段中
	s.victims = append(s.victims, victim)
	select {
	case s.victimc <- struct{}{}:   // 向 victime 远远中发送一个信号
	default:
	}
}

func (s *watchableStore) rev() int64 { return s.store.Rev() }

func (s *watchableStore) progress(w *watcher) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	// 只有 watcher 在 synced watcherGroup 中时，才会响应该方法
	if _, ok := s.synced.watchers[w]; ok {
		// 注意这里发送的 watchResponse 实例不包含任何 Event 事件，只包含当前的 revision 位
		// 如果当前 watcher.ch 通过已经阻塞，其中阻塞的 watchResponse 实例中也包含了 revision 值，
		// 也可以说明当前 watcher 的处理进皮 ，则没有必要继续添加空的 watchResponse 实例了
		w.send(WatchResponse{WatchID: w.id, Revision: s.rev()})
	}
}

// 在v3版本存储中，watcher机制支持监昕某个固定的Key值，
// 也支持监昕一个范围，这有点类似于v2版本存储中监昕一个目录节点及其子节点的效果
type watcher struct {
	key []byte	// 该watcher实例监听的原始Key值
	// 该watcher实例监昕的结束位置（也是一个原始Key值。如果该宇段有值，
	// 则当前watcher实例是一个范围watcher。如果该字段未设置值，
	// 则当前watcher只监昕上面的key字段对应的键值对。
	end []byte
	// 当ch通道阻塞时，会将该字段设置成true。
	victim bool
	// 如果该宇段被设置为true，则表示当前watcher己经因为发生了压缩操作而被删除。
	compacted bool
	restore bool
	// 能够触发当前watcher实例的最小revision值。
	// 发生在该revision之前的更新操作是无法触发该watcher实例。
	minRev int64
	id     WatchID	// 当前watcher实例的唯一标识。
	// 过滤器。触发当前wathcer实例的事件（Event结构体就是对“事件”的抽象）
	// 需要经过这些过滤器的过滤才能封装进响应
	// （WatchResponse就是对“响应”的抽象）中。
	fcs []FilterFunc
	// 当前watcher实例被触发之后，会向该通道中写入WatchResponse。
	// 该通道可能是由多个watcher实例共享的。
	ch chan<- WatchResponse
}

// sync WatchersLoop （）方 法在完成 Event 实例的转换之后， 会将 Event 实例 封装成 watchResponse 实例 ，
// 并调用 watcher.send （）方法将其写入 watcher. ch 通道中。
// 在 watcher.send() 方法中会使用 watcher.fcs 中记录的过滤器对 Event 进行过滤，
// 只有通过过滤的 Event 实例才能 被放出去，
func (w *watcher) send(wr WatchResponse) bool {
	progressEvent := len(wr.Events) == 0	// 发送 watchResponse 中封装的 Event 事件
	if len(w.fcs) != 0 {	// watcher.fcs 字段中记录了过滤. Event 实例的过滤器
		ne := make([]mvccpb.Event, 0, len(wr.Events))
		for i := range wr.Events {	// 追历 watchResponse 中封装的 Event 实例 ， 并逐个进行过滤
			filtered := false
			for _, filter := range w.fcs {
				if filter(wr.Events[i]) {
					filtered = true
					break
				}
			}
			if !filtered {	// 通过所有过滤器检测的 Event 实例才能被记录到 watchResponse 中
				ne = append(ne, wr.Events[i])
			}
		}
		wr.Events = ne
	}

	// 如果没有通过过滤的 Event 实例，则直接返回 true ，此次发送结束
	if !progressEvent && len(wr.Events) == 0 {
		return true
	}
	select {
	// 将 watchResponse 实例 写入 watcher.ch 通道中，如果watcher.ch 通道阻塞， 则通过返回 false 进行表示
	case w.ch <- wr:
		return true
	default:
		return false
	}
	// 如果watcher.ch 通道阻塞了，则会将 watchResponse 中记录的 Event 实例组织成 watch巳rBatch ，
	// 并添加到 watchableStore.victims 字段中等待处理。另外会向 watchableStore.victimc 通道中发送一个信号 ，
	// 通知其他后台 goroutine 来处理 victims 宇段中积累的 watcherBatch 实例 。
}
